package beautiful.butterfly.drds.schedule.zoo_keeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class ZooKeeperManager {

    private static transient Logger logger = LoggerFactory.getLogger(ZooKeeperManager.class);
    private ZooKeeper zooKeeper;
    private List<ACL> aclList = new ArrayList<ACL>();
    private Properties properties;
    private boolean isCheckParentPath = true;

    public enum keys {
        zkConnectString, rootPath, userName, password, zkSessionTimeout, isCheckParentPath
    }

    public ZooKeeperManager(Properties aProperties) throws Exception {
        this.properties = aProperties;
        this.connect();
    }

    /**
     * ����zookeeper
     *
     * @throws Exception
     */
    public synchronized void closeAndConnect() throws Exception {
        if (this.zooKeeper != null) {
            this.zooKeeper.close();
            this.zooKeeper = null;
            this.connect();
        }
    }

    private void connect() throws Exception {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        createZookeeper(countDownLatch);
        countDownLatch.await(10, TimeUnit.SECONDS);
    }

    private void createZookeeper(final CountDownLatch countDownLatch) throws Exception {
        zooKeeper = new ZooKeeper(this.properties.getProperty(keys.zkConnectString
                .toString()), Integer.parseInt(this.properties
                .getProperty(keys.zkSessionTimeout.toString())),
                new Watcher() {
                    public void process(WatchedEvent event) {
                        sessionEvent(countDownLatch, event);
                    }
                });
        String authString = this.properties.getProperty(keys.userName.toString())
                + ":" + this.properties.getProperty(keys.password.toString());
        this.isCheckParentPath = Boolean.parseBoolean(this.properties.getProperty(keys.isCheckParentPath.toString(), "true"));
        zooKeeper.addAuthInfo("digest", authString.getBytes());
        aclList.clear();
        aclList.add(new ACL(ZooDefs.Perms.ALL, new Id("digest",
                DigestAuthenticationProvider.generateDigest(authString))));
        aclList.add(new ACL(ZooDefs.Perms.READ, Ids.ANYONE_ID_UNSAFE));
    }

    private void sessionEvent(CountDownLatch countDownLatch, WatchedEvent watchedEvent) {
        if (watchedEvent.getState() == KeeperState.SyncConnected) {
            logger.info("�յ�ZK���ӳɹ��¼���");
            countDownLatch.countDown();
        } else if (watchedEvent.getState() == KeeperState.Expired) {
            logger.error("�Ự��ʱ���ȴ����½���ZK����...");
            try {
                closeAndConnect();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } // Disconnected��Zookeeper���Զ�����Disconnected״̬����
        else if (watchedEvent.getState() == KeeperState.Disconnected) {
            logger.info("tb_hj_schedule Disconnected���ȴ����½���ZK����...");
            try {
                closeAndConnect();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } else if (watchedEvent.getState() == KeeperState.NoSyncConnected) {
            logger.info("tb_hj_schedule NoSyncConnected���ȴ����½���ZK����...");
            try {
                closeAndConnect();
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            }
        } else {
            logger.info("tb_hj_schedule �Ự������״̬��ֵ��event.getState() =" + watchedEvent.getState() + ", event  value=" + watchedEvent.toString());
            countDownLatch.countDown();
        }
    }

    public void close() throws InterruptedException {
        logger.info("�ر�zookeeper����");
        if (zooKeeper == null) {
            return;
        }
        this.zooKeeper.close();
    }

    public static Properties createProperties() {
        Properties properties = new Properties();
        properties.setProperty(keys.zkConnectString.toString(), "localhost:2181");
        properties.setProperty(keys.rootPath.toString(), "/taobao-pamirs-schedule/huijin");
        properties.setProperty(keys.userName.toString(), "ScheduleAdmin");
        properties.setProperty(keys.password.toString(), "password");
        properties.setProperty(keys.zkSessionTimeout.toString(), "60000");
        properties.setProperty(keys.isCheckParentPath.toString(), "true");

        return properties;
    }

    public String getRootPath() {
        return this.properties.getProperty(keys.rootPath.toString());
    }

    public String getConnectStr() {
        return this.properties.getProperty(keys.zkConnectString.toString());
    }

    public boolean checkZookeeperState() throws Exception {
        return zooKeeper != null && zooKeeper.getState() == States.CONNECTED;
    }

    public void initial() throws Exception {
        //��zoo_keeper״̬��������ܵ���
        if (zooKeeper.exists(this.getRootPath(), false) == null) {
            ZooKeeperTools.createPath(zooKeeper, this.getRootPath(), CreateMode.PERSISTENT, aclList);
            if (isCheckParentPath == true) {
                checkParent(zooKeeper, this.getRootPath());
            }
            //���ð汾��Ϣ
            zooKeeper.setData(this.getRootPath(), Version.getVersion().getBytes(), -1);
        } else {
            //��У�鸸�׽ڵ㣬�����Ƿ��Ѿ���schedule��Ŀ¼
            if (isCheckParentPath == true) {
                checkParent(zooKeeper, this.getRootPath());
            }
            byte[] value = zooKeeper.getData(this.getRootPath(), false, null);
            if (value == null) {
                zooKeeper.setData(this.getRootPath(), Version.getVersion().getBytes(), -1);
            } else {
                String dataVersion = new String(value);
                if (Version.isCompatible(dataVersion) == false) {
                    throw new Exception("TBSchedule����汾 " + Version.getVersion() + " ������Zookeeper�е����ݰ汾 " + dataVersion);
                }
                logger.info("��ǰ�ĳ���汾:" + Version.getVersion() + " ���ݰ汾: " + dataVersion);
            }
        }
    }

    public static void checkParent(ZooKeeper zk, String path) throws Exception {
        String[] list = path.split("/");
        String zkPath = "";
        for (int i = 0; i < list.length - 1; i++) {
            String str = list[i];
            if (str.equals("") == false) {
                zkPath = zkPath + "/" + str;
                if (zk.exists(zkPath, false) != null) {
                    byte[] value = zk.getData(zkPath, false, null);
                    if (value != null) {
                        String tmpVersion = new String(value);
                        if (tmpVersion.indexOf("taobao-pamirs-schedule-") >= 0) {
                            throw new Exception("\"" + zkPath + "\"  is already a schedule instance's root directory, its any subdirectory cannot as the root directory of others");
                        }
                    }
                }
            }
        }
    }

    public List<ACL> getAclList() {
        return aclList;
    }

    public ZooKeeper getZooKeeper() throws Exception {
        if (this.checkZookeeperState() == false) {
            closeAndConnect();
        }
        return this.zooKeeper;
    }

}
